package cn.edu.gmall.realtime.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class MyKafkaUtil {
    private static final Properties properties = new Properties();

    static {
        properties.setProperty("bootstrap.servers", "localhost:9092");
    }

    public static FlinkKafkaProducer<String> getKafkaSink(String topic) {
        return new FlinkKafkaProducer<String>(topic, new SimpleStringSchema(), properties);
    }


    public static FlinkKafkaConsumer<String> getKafkaSource(String topic, String groupId) {
        //给配置信息对象添加配置项
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.setProperty("auto.offset.reset", "earliest");

        //获取 KafkaSource
        return new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), properties);
    }

    public static <T> FlinkKafkaProducer<T> getKafkaSinkBySchema(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        String DEFAULT_TOPIC = "dwd_default_topic";
        return new FlinkKafkaProducer<T>(DEFAULT_TOPIC, kafkaSerializationSchema, properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    //拼接 Kafka 相关属性到 DDL
    public static String getKafkaDDL(String topic, String groupId) {
        String ddl = "" +
                "'connector' = 'kafka'," +
                "'topic' = '" + topic + "'," +
                "'properties.bootstrap.servers' = 'localhost:9092'," +
                "'properties.group.id' = '" + groupId + "'," +
                "'format' = 'json'," +
                "'scan.startup.mode' = 'earliest-offset'";
        return ddl;
    }
}
